/*
主要处理逻辑，从kafka中读取日志信息，发送给ES
*/
package mainfuncs

import (
	"fmt"

	"github.com/Shopify/sarama"
	"github.com/astaxie/beego/logs"
)

func Run() (err error) {

	partitionList, err := AppKafkaClient.client.Partitions(AppKafkaClient.topic)
	if err != nil {
		panic(err)
	}

	for partition := range partitionList {
		pc, err := AppKafkaClient.client.ConsumePartition(AppKafkaClient.topic, int32(partition), sarama.OffsetNewest)
		if err != nil {
			panic(err)
		}

		defer pc.AsyncClose()

		AppKafkaClient.wg.Add(1)

		go func(sarama.PartitionConsumer) {
			defer AppKafkaClient.wg.Done()

			for msg := range pc.Messages() {
				fmt.Printf("Partition:%d, Offset:%d, Key:%s, Value:%s\n", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
				logs.Debug("Partition:%d, Offset:%d, Key:%s, Value:%s", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
				//fmt.Println()
				err = sendToES(AppKafkaClient.topic, msg.Value) //发送日志消息到ES中
				if err != nil {
					logs.Warn("send to es failed, err:%v", err)
				}
			}

		}(pc)
	}
	AppKafkaClient.wg.Wait()
	AppKafkaClient.client.Close()

	return
}
